/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.trp.report.merge;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;

import com.semidata.rtd.core.log.Logger;
import com.semidata.trp.report.config.Config;
import com.semidata.trp.report.data.UserData;
import com.semidata.trp.report.sort.FileDataProvider;
import com.semidata.trp.report.utils.SqlUtils;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class MonitorDataMerge {

	private Config config = Config.getConfig();

	private char[] b64 = {'+', '-', '0', '1', '2', '3', '4', '5', '6', '7',
	        '8', '9', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K',
	        'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X',
	        'Y', 'Z', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k',
	        'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x',
	        'y', 'z'};

	public MonitorDataMerge() {

	}

	private static class ParseResult {
		private String userId;
		private int resourceId;
		private int strategyId;
		private int imp;
		private int click;
		private int cost;

		public ParseResult(String userId, int resourceId, int strategyId,
		        int imp, int click, int cost) {
			this.userId = userId;
			this.resourceId = resourceId;
			this.strategyId = strategyId;
			this.imp = imp;
			this.click = click;
			this.cost = cost;
		}

		public String getUserId() {
			return userId;
		}

		public int getResourceId() {
			return resourceId;
		}

		public int getStrategyId() {
			return strategyId;
		}

		public int getImp() {
			return imp;
		}

		public int getClick() {
			return click;
		}

		public int getCost() {
			return cost;
		}

	}

	private File getTmpDir() {

		File dir = new File(config.getReportDirTmp());
		if (dir.exists()) {
			for (File file : dir.listFiles()) {
				file.delete();
			}
			dir.delete();
		}

		if (!dir.mkdirs()) {
			throw new RuntimeException("bad tmp dir, please check");
		}

		return dir;
	}

	private void completeMerge(File tmpDir, File destDir) {

		File parent = destDir.getParentFile();
		if (!parent.exists()) {
			if (!parent.mkdirs()) {
				throw new RuntimeException("bad dir, please check");
			}
		}

		try {
			Files.move(tmpDir.toPath(), destDir.toPath(),
			        StandardCopyOption.ATOMIC_MOVE);
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException("bad dir, please check");
		}
	}

	public void merge(final String minute, final boolean force) {

		if (minute == null || minute.length() != 12)
			throw new RuntimeException(
			        "bad argument: " + minute == null ? "null" : minute);

		List<String> servers;
		try {
			servers = SqlUtils.getMonitorServerNames();
		} catch (Exception e) {
			return;
		}
		
		File dir = new File(config.getReportDirMonitor());
		dir = new File(dir, minute.substring(0, 8));

		final List<File> fileList = new ArrayList<>();

		final boolean complete = getFileList(dir, minute, fileList, servers);

		if (!complete && !force) {
			return;
		}

		File destDir = getTmpDir();

		final MonitorMergeThread[] threads = new MonitorMergeThread[config
		        .getThreads()];
		final Map<Character, MonitorMergeThread> threadMap = new HashMap<>();
		ArrayList<List<Character>> suffixes = new ArrayList<>();
		for (int i = 0; i < config.getThreads(); i++) {
			suffixes.add(new ArrayList<Character>());
		}
		for (int i = 0; i < 64; i++) {
			suffixes.get(i % config.getThreads()).add(b64[i]);
		}
		for (int i = 0; i < config.getThreads(); i++) {
			threads[i] = new MonitorMergeThread(suffixes.get(i), destDir);
			threads[i].merge();
		}

		for (int i = 0; i < 64; i++) {
			threadMap.put(b64[i], threads[i % config.getThreads()]);
		}

		Thread thread = new Thread(new Runnable() {

			@Override
			public void run() {

				for (File file : fileList) {

					FileDataProvider provider = new FileDataProvider(file);

					String line;

					while ((line = provider.poll()) != null) {
						try {
							ParseResult data = parse(line);
							if (data != null) {
								UserData userData = new UserData();
								userData.setUserId(data.getUserId());
								userData.add(data.getResourceId(),
								        data.getStrategyId(), data.getImp(),
								        data.getClick(), data.getCost());
								char ch = data.getUserId().charAt(0);
								MonitorMergeThread subThread = threadMap
								        .get(ch);
								subThread.add(userData);
							} else {
								Logger.log("DEBUG", "can not parse line: "
								        + line);
							}
						} catch (Exception e) {
							Logger.log("DEBUG", "can not parse line: " + line);
						}
					}
				}
				for (MonitorMergeThread subThread : threads) {
					subThread.quit();
				}
			}
		});

		thread.start();

		try {
			thread.join();
		} catch (InterruptedException e) {
		}

		for (MonitorMergeThread subThread : threads) {
			subThread.join();
		}

		try {
			FileOutputStream out = new FileOutputStream(new File(destDir,
			        complete ? "COMPLETE" : "INCOMPLETE"));
			out.close();
		} catch (Exception e) {
			e.printStackTrace();
		}

		File finalDir = new File(config.getReportDirMinute());
		finalDir = new File(finalDir, minute.substring(0, 6));
		finalDir = new File(finalDir, minute.substring(0, 8));
		finalDir = new File(finalDir, minute.substring(0, 10));
		finalDir = new File(finalDir, minute);
		completeMerge(destDir, finalDir);
	}

	private boolean getFileList(File dir, String minute, List<File> list,
	        List<String> servers) {

		File[] allList = dir.listFiles();
		Set<String> set = new HashSet<>();
		Set<File> results = new TreeSet<>();
		boolean complete = true;

		if (allList == null || allList.length == 0) {
			return false;
		}

		String prefix = minute + ".";
		for (File file : allList) {
			if (file.getName().startsWith(prefix))
				results.add(file);
			set.add(file.getAbsolutePath());
		}

		if (servers == null)
			return false;

		for (String server : servers) {
			String logFile = minute + "." + server + ".log";
			File file = new File(dir, logFile);
			if (!set.contains(file.getAbsolutePath())) {
				complete = false;
			} else {
				results.add(file);
			}
		}

		for (File file : results) {
			list.add(file);
		}
		return complete;
	}

	private ParseResult parse(String line) {

		int imp = 0;
		int click = 0;
		int cost = 0;
		String userId;
		int resource;
		int strategy;
		if (line.startsWith("/a.gif?")) {
			imp = 1;
		} else if (line.startsWith("/r.gif?")) {
			click = 1;
		} else {
			return null;
		}

		try {
			userId = getValue(line, "u");
			resource = Integer.parseInt(getValue(line, "r"));
			strategy = Integer.parseInt(getValue(line, "s"));
			if (imp == 1)
				cost = (int) (Double.parseDouble(getValue(line, "b")) * 1000);
			else
				cost = 0;
		} catch (Exception e) {
			return null;
		}

		return new ParseResult(userId, resource, strategy, imp, click, cost);
	}

	private String getValue(String line, String prefix) throws Exception {
		int pos = line.indexOf(prefix + "=");
		if (pos == -1)
			throw new Exception();
		int start = pos + 2;
		pos = line.indexOf("&", start);
		return line.substring(start, pos);
	}

	// static {
	// Config.init(null);
	// }
	//
	// public static void main(String[] args) {
	// MonitorDataMerge merge = new MonitorDataMerge();
	// merge.merge("201407260207", true);
	// }

}
